package com.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @describte:kafka消费者
 * @program:foodie-dev
 * @author:zhengdp
 * @create:2021-09-18 11:47:46
 **/
@Slf4j
public class CollectKafkaConsumer {

    // 消费者实例
    public final KafkaConsumer<String,String> consumer;
    // 消费主题
    private final String topic;

    public CollectKafkaConsumer(String topic){
        // 配置 Consumer配置
        Properties props = new Properties();
        // 配置 zookeeper地址
        props.put("zookeeper.connect","192.168.206.131:2181");
        // 配置 broker地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.206.70:9092");
        // 定义消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"demo-group-id");
        // 是否自动提交（auto commit，一般生产环境均设置为false，则为手工确认）
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 自动提交配置项
        // props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 消费进度（位置 offset）重要设置: latest,earliest
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 超时时间配置
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        // kafka序列化配置
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建 Consumer对象
        consumer = new KafkaConsumer<String, String>(props);
        this.topic = topic;
        // Consumer对象订阅主题  ===> 可以订阅多个主题
        consumer.subscribe(Collections.singletonList(topic));

    }

    /**
     * 循环拉取消息并进行消费， 使用手工ACK方式
     * @param consumer 消费者
     */
    public void receive(KafkaConsumer<String,String> consumer){
        while(true){
            // 拉取结果集 (拉取超时时间为1秒）
            ConsumerRecords<String,String> records = consumer.poll(Duration.ofSeconds(1));
            // 拉取结果集后获取到 遍历结果集中的所有分区
            for(TopicPartition partition : records.partitions()){
                // 获取到对应分区中的结果集
                List<ConsumerRecord<String, String>> partitionRecords  = records.records(partition);
                // 获取到该分区对应的topic
                String topic = partition.topic();
                // 获取该分区返回的消息数量
                int size = partitionRecords.size();
                log.info("获取topic:{},分区位置:{},消息数为:{}", topic, partition.partition(), size);
                // 处理各个分区中的消息
                for(int i = 0;i<size;i++){
                    // 打印消息内容
                    System.err.println("-----> value: " + partitionRecords.get(i).value());
                    // 获取到当前消息的offset,利用该offset 进行手动ACK
                    long offset = partitionRecords.get(i).offset() + 1;
                    // consumer.commitSync(); // 这种提交会自动获取partition 和 offset
                    // 这种是显示提交partition 和 offset 进度
                    consumer.commitSync(Collections.singletonMap(partition,
                            new OffsetAndMetadata(offset)));
                    log.info("同步成功, topic: {}, 提交的 offset: {} ", topic, offset);
                }
            }
        }
    }
}
